package cn.ipanel.bigdata.job

import cn.ipanel.bigdata.boot.period.Period
import cn.ipanel.bigdata.dw.dwm.AbsDwmHive
import cn.ipanel.bigdata.dw.dws.{AbsDwsPhoenix, parseDateToPhoenixDate}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.{col, countDistinct, last, lit, sum, sumDistinct, when}

/**
 * Author: lzz
 * Date: 2022/6/22 17:11
 */
package object dws {

  abstract class AbsDwsJob extends AbsJob {

    def optSourceTable: AbsDwmHive
    lazy val _S: AbsDwmHive = optSourceTable

    def optTargetTable: AbsDwsPhoenix
    lazy val _T: AbsDwsPhoenix = optTargetTable

    private[this] lazy val _PERIOD: Seq[Period] = {
//      getTemplate.asPeriod ++ Seq(Period.DAY, Period.WEEK, Period.MONTH, Period.QUARTER, Period.YEAR)
      // 陕西现场只要天周期
      getTemplate.asPeriod ++ Seq(Period.DAY)
    }

    /**
     * 可选观测视角，每个实现了 {@link AbsDwsJob } 的任务定制化自己的视角
     * @return
     */
    def optExtraAngle: Seq[String] = Seq()

    /**
     * 所有观测视角，包括：
     * 1.外部配置的可选观测视角（以任务名称来配置，所以也是可选视角）
     * 2.固定观测视角
     * 3.所有业务应该定死的必须包含的观测视角，目前是 区域，产商，拍照上，周期
     */
    private[this] lazy val _ANGLE: Seq[String] = {
      getTemplate.asAngle ++
        optExtraAngle ++
        Seq(_T.F_PERIOD_TIME, _T.F_REGION)
    }

    /**
     * 用于聚合的栏目字段，必须有对次数和设备数
     */
    private[this] lazy val _AGG_COLUMNS: Seq[org.apache.spark.sql.Column] = {
      var cols: Seq[org.apache.spark.sql.Column] = Seq()
      _T.getTBColumns.foreach(colName => {
        if (!_ANGLE.contains(colName) &&
          !_T.F_COUNT.equals(colName) &&
          !_T.F_DEVICE_COUNT.equals(colName) &&
          !isAggColumn(colName))
          cols ++= Seq(last(colName).isNotNull as colName)
      })
      cols ++ optAggColumn ++ Seq(sum(_T.F_COUNT) as _T.F_COUNT,
                  countDistinct(_S.F_DEVICE_ID) as _T.F_DEVICE_COUNT)
    }

    /**
     * 可选的聚合字段聚合方式，为除了固定聚合方式（目前固定方式只有按周期的次数聚合）以外的其他聚合
     * @return
     */
    def optAggColumn: Seq[org.apache.spark.sql.Column] = Seq()

    private[this] def isAggColumn(columnName: String): Boolean = {
      for (elem <- optAggColumn) {
        // like  sum(AA) AS `AA`，左边的是hive的字段，右边的是hbase的字段，二者可能相等
        // like  count(DISTINCT CASE WHEN (f_cpu_exceed_count = 0) THEN NULL ELSE f_device_id END) AS `f_cpu_exceed_device_count`
        var tmp = elem.toString()
        val strings = tmp.split(" AS ")

        if (strings(0).contains(columnName) || strings(1).contains(columnName)){
          return true
        }

      }

      false
    }

    // hive 参与计算，一定要排除的数据（多了也可以，但是一定没必要）
    protected final val withoutColumns = Seq(_S.F_DATE, _S.F_CREATE_TIME, _S.F_MODIFY_TIME, _S.F_PERIOD_TYPE)
    override def source(): DataFrame = _S.find(_DAY).drop(withoutColumns: _*)
    override def onPrepare(): Unit = _PERIOD.foreach(_T.drop(_DAY, _))

    override def onStartup(): Unit = {
      println("optAggColumn: " + optAggColumn)
      println("_AGG_COLUMNS: " + _AGG_COLUMNS)

      var src: DataFrame = null
      _PERIOD.foreach((period: Period) => {
        if(src == null) {
          src = source()
            .withColumn(_T.F_PERIOD_TIME, lit(parseDateToPhoenixDate(_DAY, period)))
        } else {
          // 获取统计天所属周期的Target数据，跟天合并，计算包含该天的Target
          src = src.union(
            _S.find(_DAY, period)
              .drop(withoutColumns: _*)
              .withColumn(_T.F_PERIOD_TIME, lit(parseDateToPhoenixDate(_DAY, period)))
          )
        }
      })

      // 通过 hive 明细算无用户的汇总
      _T.save({
        src.groupBy(_ANGLE.head, _ANGLE.tail: _*)
           .agg(_AGG_COLUMNS.head, _AGG_COLUMNS.tail: _*)
           .selectExpr(_T.getTBColumns: _*)
      })
      src.unpersist()
    }

  }
}